package io.github.junxworks.junx.stream.pipeline;

import com.google.common.collect.Lists;
import io.github.junxworks.junx.core.lang.Initializable;
import io.github.junxworks.junx.core.util.ClockUtils;
import io.github.junxworks.junx.event.EventChannel;
import io.github.junxworks.junx.stream.constants.ComponentStatus;
import io.github.junxworks.junx.stream.handler.AbstractDataHandler;
import lombok.Getter;
import lombok.Setter;

import java.util.List;
import java.util.function.Consumer;

/**
 * 管道中的基础组件
 *
 * @author 王兴
 * @date 2024/09/10
 */
@Getter
@Setter
public abstract class BaseComponent implements Initializable {
    /**
     * 流程名
     */
    protected String flowName;
    protected ClockUtils.Clock clock = ClockUtils.createClock();
    /**
     * 当前步骤名
     */
    protected String stepName;
    protected ComponentStatus status = ComponentStatus.STOPPED;
    protected List<BaseComponent> children = Lists.newArrayList();
    protected BaseComponent parent;
    /**
     * 数据管道
     */
    protected EventChannel eventChannel;

    protected AbstractDataHandler eventChannelHandler;

    /**
     * 从哪个数据主题接收数据
     */
    protected String inputTopic;
    /**
     * 从本组件流出的数据主题
     */
    protected String outputTopic;

    public int bufferSize;

    public void addChild(BaseComponent child) {
        children.add(child);
    }

    public void removeChild(BaseComponent child) {
        children.remove(child);
    }

    public boolean isDone(){
        return eventChannelHandler.isDone();
    }

    public void stop() {
        this.status = ComponentStatus.STOPPED;
    }

    public void start() {
        this.status = ComponentStatus.RUNNING;
    }

    public void waitFor() {
        this.status = ComponentStatus.WAITING;
    }

    public void accept(Consumer<BaseComponent> consumer) {
        consumer.accept(this);
        children.forEach(child -> child.accept(consumer));
    }

}
